package com.iqiyi.pps.epg.web.timer;

import com.iqiyi.pps.epg.core.model.channel.ChannelBase;
import com.iqiyi.pps.epg.core.model.channel.ChannelConfig;
import com.iqiyi.pps.epg.core.model.stream.LiveChannelStream;
import com.iqiyi.pps.epg.core.service.channel.ChannelService;
import com.iqiyi.pps.epg.core.service.stream.LiveChannelStreamService;

import com.iqiyi.pps.epg.core.utils.SendMailUtils;
import org.dom4j.Document;
import org.dom4j.DocumentException;
import org.dom4j.Element;
import org.dom4j.io.SAXReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 直播流同步任务 (deprecated)
 * 每3分分钟检测下所有直播流是否active，并更新码流等信息。否者报警
 */
public class ChannelStreamSyncTrigger {
    private static Logger logger = LoggerFactory.getLogger(ChannelStreamSyncTrigger.class);

    @Autowired
    private LiveChannelStreamService liveChannelStreamService;
    @Autowired
    private ChannelService channelService;

    private ConcurrentLinkedQueue<LiveChannelStream> queue;
    private List<LiveChannelStream> idleStream = new ArrayList();

    public void task() {
        logger.info("[ChannelStreamSyncTrigger][task][start]");
        List<LiveChannelStream> results = liveChannelStreamService.getAll();
        if (results.size() > 0) {
            queue = new ConcurrentLinkedQueue<LiveChannelStream>(results);
            logger.info("[ChannelStreamSyncTrigger][task][ConcurrentLinkedQueue size={}]", queue.size());
            ExecutorService exec = Executors.newFixedThreadPool(5);
            for (int i = 0; i < 5; i++) {
                exec.execute(new Executor());
            }
            exec.shutdown();
            try {
                while (!exec.awaitTermination(5, TimeUnit.SECONDS));
            } catch (InterruptedException e) {
                logger.info("[ChannelStreamSyncTrigger][task][awaitTermination][exception={}]", e);
            }
            // 发送邮件
            if (idleStream.size() > 0) {
                StringBuilder sb = new StringBuilder("以下流未编码状态<br /><br />");
                for (LiveChannelStream item : idleStream) {
                    sb.append("<a href='http://")
                        .append(item.getStreamIp())
                        .append(":8080/stat' target='_blank'>rtmp://")
                        .append(item.getStreamIp())
                        .append("/live/")
                        .append(item.getStreamName())
                        .append("</a><br />");
                }
                idleStream.clear(); // 清空本次数据
                List<ChannelConfig> mailList = channelService.getConfig(ChannelConfig.TEAM_TRIGGER_MAIL_LIST);
                String[] tos = new String[mailList.size()];
                int i = 0;
                for (ChannelConfig item : mailList) {
                    tos[i] = item.getVal();
                    ++i;
                }

                SendMailUtils.send("直播流同步任务报警", tos, sb.toString());
            }
        }
    }

    public class Executor implements Runnable {
        @Override
        public void run() {
            LiveChannelStream entity;
            while (true) {
                entity = null;
                synchronized (queue) {
                    if (!queue.isEmpty()) {
                        entity = queue.poll();
                        logger.info("[ChannelStreamSyncTrigger][thread][id={}]", entity.getId());
                    }
                }

                // 无任务退出
                if (entity == null) {
                    logger.info("[ChannelStreamSyncTrigger][thread][exit]");
                    break;
                }

                // 删除状态不处理
                if (entity.getStatus() == LiveChannelStream.STATUS_DELETE) {
                    logger.info("[ChannelStreamSyncTrigger][thread][delete id={}]", entity.getId());
                    continue;
                }

                // http://10.121.33.71:8080/stat
                String url = "http://" + entity.getStreamIp() + ":8080/stat";
                logger.info("[ChannelStreamSyncTrigger][thread][while][id={}][url={}]", entity.getId(), url);
                try {
                    SAXReader saxReader = new SAXReader();
                    Document document = saxReader.read(url);
                    Element rootElement = document.getRootElement();
                    List nodes = rootElement.selectNodes("//server/application/live/stream[name='" + entity.getStreamName() + "']");
                    if (nodes.size() > 0) {
                        Element node = (Element) nodes.get(0);
                        boolean actived = false;
                        boolean updated = false;
                        for (Iterator i = node.elementIterator(); i.hasNext(); ) {
                            Element ele = (Element) i.next();
                            switch (Node.toNode(ele.getName().toUpperCase())) {
                                case BWIN:
                                    if (!ele.getText().equals(entity.getInputBits())) {
                                        entity.setInputBits(ele.getText());
                                        updated = true;
                                    }
                                    break;
                                case BWOUT:
                                    if (!ele.getText().equals(entity.getOutputBits())) {
                                        entity.setOutputBits(ele.getText());
                                        updated = true;
                                    }
                                    break;
                                case META:
                                    int width = 0, height = 0;
                                    for (Iterator j = ele.elementIterator(); j.hasNext(); ) {
                                        Element item = (Element) j.next();
                                        switch (Node.toNode(item.getName().toUpperCase())) {
                                            case WIDTH:
                                                width = Integer.valueOf(item.getText());
                                                break;
                                            case HEIGHT:
                                                height = Integer.valueOf(item.getText());
                                                break;
                                            case FRAMERATE:
                                                if (!item.getText().equals(entity.getFrameRate())) {
                                                    entity.setFrameRate(item.getText());
                                                    updated = true;
                                                }
                                                break;
                                            case BITRATE:
                                                if (!item.getText().equals(entity.getBitRate().toString())) {
                                                    entity.setBitRate(Long.valueOf(item.getText()));
                                                    updated = true;
                                                }
                                                break;
                                            case VIDEO:
                                                if (!item.getText().equals(entity.getVideo())) {
                                                    entity.setVideo(item.getText());
                                                    updated = true;
                                                }
                                                break;
                                            case AUDIO:
                                                if (!item.getText().equals(entity.getAudio())) {
                                                    entity.setAudio(item.getText());
                                                    updated = true;
                                                }
                                                break;
                                        }
                                    }
                                    if (width > 0 && height > 0) {
                                        String size = width + ":" + height;
                                        if (!size.equals(entity.getScreenSize())) {
                                            entity.setScreenSize(size);
                                            // 画面比例
                                            double scale = (double) (width / height);
                                            if (scale >= 1.5) {
                                                entity.setAspectRatio("16:9");
                                            } else if (scale <= 1.33) {
                                                entity.setAspectRatio("4:3");
                                            } else {
                                                entity.setAspectRatio("");
                                            }
                                            updated = true;
                                        }
                                    }
                                    break;
                                case ACTIVE:
                                    actived = true;
                                    if (entity.getStatus() != LiveChannelStream.STATUS_ACTIVED) {
                                        entity.setStatus(LiveChannelStream.STATUS_ACTIVED);
                                        updated = true;
                                    }
                                    break;
                            }
                        }

                        if (!actived) {
                            if (entity.getStatus() != LiveChannelStream.STATUS_IDLED) {
                                entity.setStatus(LiveChannelStream.STATUS_IDLED);
                                updated = true;
                            }
                            idleStream.add(entity); // 报警邮件
                            // 更新频道启用状态为无效
                            ChannelBase channelInfo = channelService.getChannelBase(entity.getFatherId());
                            channelInfo.setIsEnable(ChannelBase.DISABLE);
                            channelService.saveBase(channelInfo);
                        }
                        // 更新数据
                        if (updated) {
                            liveChannelStreamService.save(entity);
                        }
                    } else {
                        logger.info("[ChannelStreamSyncTrigger][thread][notfound][id={}][streamName={}]", entity.getId(), entity.getStreamName());
                    }
                } catch (DocumentException e) {
                    logger.info("[ChannelStreamSyncTrigger][thread][id={}][url={}][exception={}]", entity.getId(), url, e);
                } catch (Exception e) {
                    logger.info("[ChannelStreamSyncTrigger][thread][id={}][exception={}]", entity.getId(), e);
                }
                logger.info("[ChannelStreamSyncTrigger][thread][while][id={}][end]", entity.getId());
            }
        }
    }

    // 1.7 switch 支持 string
    public enum Node {
        ACTIVE, BWIN, BWOUT, META, BITRATE, FRAMERATE, WIDTH, HEIGHT, VIDEO, AUDIO,
        NOVALUE;

        public static Node toNode(String node) {
            try {
                return valueOf(node);
            } catch (Exception e) {
                return NOVALUE;
            }
        }
    }
}
